package com.shu.es;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;

public class EsMoveData {

	public static void main(String[] args) throws Exception {
		String index = args[0];
		 //备份集群
		RestHighLevelClient backupClient = getClient("127.0.0.1",9201);
		RestHighLevelClient client = getClient("127.0.0.1",9200); //主集群
		
		BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		searchSourceBuilder.query(boolQuery);
		String[] indes = {index};
		SearchRequest searchRequest = new SearchRequest(indes,searchSourceBuilder);
		searchSourceBuilder.from(130001);
		searchSourceBuilder.size(60000);
		SearchResponse response = backupClient.search(searchRequest);
		SearchHit[] hits = response.getHits().getHits();
		Long size = response.getHits().getTotalHits();
		System.out.println("========================================>"+response.getHits().getTotalHits());
		BulkRequest request = new BulkRequest(); 
		int i = 1;
		for(SearchHit hit: hits) {
			Map<String, Object> map = hits[0].getSourceAsMap();
			//System.out.println(map);
			request.add(new IndexRequest(index, "data",hits[0].getId()).source(map));
			System.out.println(size + ":" + i + ":" + hits[0].getId());
			i++;
		}
		BulkResponse bulk = client.bulk(request);
		System.out.println(bulk.status());
		System.out.println(bulk.buildFailureMessage());
		backupClient.close();
		client.close();
	}
	
	
	public static RestHighLevelClient getClient(String ip,Integer port) {
		List<HttpHost> list = new ArrayList<HttpHost>();
		list.add(new HttpHost(ip, port, "http"));
		HttpHost[] arr = new HttpHost[list.size()];
		arr = list.toArray(arr);
		
		RestClientBuilder builder = RestClient.builder(arr)
				.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
					public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
						requestConfigBuilder.setConnectTimeout(5000);
						requestConfigBuilder.setSocketTimeout(40000);
						requestConfigBuilder.setConnectionRequestTimeout(1000);
						return requestConfigBuilder;
					}

				}).setMaxRetryTimeoutMillis(20 * 60 * 1000);

		return new RestHighLevelClient(builder);
	}
}
